package org.example;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 流处理
 */
public class StreamWordCount {
    public static void main(String[] args) throws Exception {
        //流处理执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //env.setParallelism(1);//并行度

        ParameterTool parameterTool = ParameterTool.fromArgs(args);
        String host = parameterTool.get("host");
        int port = parameterTool.getInt("port");

        DataStream<String> dataStream = env.socketTextStream(host, port);

        DataStream<Tuple2<String, Integer>> resultStream = dataStream.flatMap((FlatMapFunction<String, Tuple2<String, Integer>>) (s, collector) -> {
            String[] words = s.split(" ");
            for (String word : words) {
                collector.collect(new Tuple2<>(word, 1));
            }

        }).returns(Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(e -> e.f0)
                .sum(1);


        resultStream.print();

        env.execute();//执行任务

    }
}
// org.example.StreamWordCount
// --host 192.168.228.1 --port 7777


